package com.baiyan.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baiyan.model.OrderDTO;
import com.baiyan.util.GsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.Nullable;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * 发送topic测试数据值topic
 *
 * @author baiyan
 * @date 2022/01/19
 */
@RestController
@RequestMapping(value = "/kafka-sort-consume-master")
@Slf4j
public class OrderController {
    static String BATCH_TOPIC_TEST="big_data_topic";
    @Autowired
    KafkaTemplate kafkaTemplate;

    @GetMapping("/send")
    public Object send() {
        List<OrderDTO> orderDTOList = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            OrderDTO order = new OrderDTO();
            order.setOrderName("订单" + UUID.randomUUID());
            order.setId(Long.valueOf(i));
            order.setStatus(i);
            //orderDTOList.add(order);
            ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(BATCH_TOPIC_TEST, JSONObject.toJSONString(order));
            LocalDateTime dateTime = LocalDateTime.now();
            log.info("topic：{}发送消息完成,完成时间：{}",BATCH_TOPIC_TEST,dateTime);
            send.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    log.error(BATCH_TOPIC_TEST+"-生产者发送消息失败"+throwable.getMessage());
                }

                @Override
                public void onSuccess(@Nullable SendResult<String, String> stringObjectSendResult) {
                    LocalDateTime now = LocalDateTime.now(); // 获取当前时间
                    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); // 定义日期格式化器
                    String dateTime = now.format(formatter);
                    log.info(BATCH_TOPIC_TEST+"-生产者发送消息成功"+stringObjectSendResult.toString()+"时间："+dateTime);
                }
            });
             kafkaTemplate.send("big_data_topic", GsonUtil.gsonToString(order));
        }
        //kafkaTemplate.send("big_data_topic", GsonUtil.gsonToString(orderDTOList));
        return "ok";
    }
   /* @GetMapping("/send2")
    public Object send2() {
        List<String> orderDTOList = new ArrayList<>();
        for (long i = 0; i < 5; i++) {
            OrderDTO order = new OrderDTO();
            order.setOrderName("订单" + i);
            order.setId(i);
            orderDTOList.add(JSON.toJSONString(order));
            //kafkaTemplate.send("big_data_topic", GsonUtil.gsonToString(order));
        }
      //  kafkaTemplate.send("big_data_topic", orderDTOList);

        List<ProducerRecord<Object, String>> records = orderDTOList.stream()
                .map(obj -> new ProducerRecord<>("my-big_data_topic", obj))
                .collect(Collectors.toList());
        kafkaTemplate.send("big_data_topic", records);
        return "ok";
    }*/
}
